package com.leeyu.rabbitmq.config;

import com.leeyu.mapper.ResendMapper;
import com.leeyu.util.GlobalCache;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.Map;
import java.util.Objects;

@Slf4j
@Configuration
public class RabbitMQTemplateConfig {

    @Autowired
    private ResendMapper resendMapper;

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setConfirmCallback(confirmCallback());
        rabbitTemplate.setReturnCallback(returnCallback());
        rabbitTemplate.setMandatory(Boolean.TRUE);
        return rabbitTemplate;
    }

    private RabbitTemplate.ConfirmCallback confirmCallback(){
        return new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                if (Objects.isNull(correlationData)) return;
                String correlationDataId = correlationData.getId();
                if (ack) {
                    log.info("消息成功发送到Exchange, correlationDataId={}", correlationDataId);
                } else {
                    log.error("消息发送到Exchange失败, correlationDataId={}", correlationDataId);
                    resendMapper.saveMessage((Map<String, String>) GlobalCache.get(correlationDataId));
                }
                GlobalCache.remove(correlationDataId);
            }
        };
    }

    private RabbitTemplate.ReturnCallback returnCallback(){
        return new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                log.error("消息路由到队列失败！");
                log.error("消息：{}", new String(message.getBody()) + ", replyCode=" + replyCode + ", replyText=" + replyText);
                log.error("交换机：{}", exchange);
                log.error("路由键：{}", routingKey);
                // 入库等待重试......
            }
        };
    }
}
